package com.niit.spark.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Date:2025/4/15
 * Author：Ys
 * Description:
 */
object Action_RDD extends App {

  //准备环境
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Action_RDD")
  val sc = new SparkContext(sparkConf)
  sc.setLogLevel("ERROR") // 设置日志打印的级别 Error

   /*
    行动算子
     reduce：聚合RDD中的所有元素，先聚合分区内，再聚合分区间，返回一个值
    */
  val rdd1 = sc.makeRDD( List(1,2,3,4) , 2 ) // 【1，2】 = 3   【3，4】 = 7   3 + 7 = 10
  val i: Int = rdd1.reduce(_ + _)
  println(i)


  /*
    collect：将RDD中的数据收集到Driver端内存中，返回一个Array
        将RDD转换成数组
    */
   val arrs1: Array[Int] = rdd1.collect()
  println(arrs1.mkString(","))

  /*
  count: 返回RDD中元素的个数
   */
  val co: Long = rdd1.count()
  println(co)

  /*
    first：返回RDD中的第一个元素
     */
  val first: Int = rdd1.first()
  println("first:",first)

  /*
    take：返回RDD中前n个元素，返回一个Array
   */
  val take: Array[Int] = rdd1.take(2)
  println("take:",take.mkString(","))


  /*
    takeOrdered：返回RDD中前n个元素，返回一个Array，默认升序
   */
   val rdd2: RDD[Int] = sc.makeRDD(List(3, 1, 4, 2))
   val arr3: Array[Int] = rdd2.takeOrdered(2)
   println("takeOrdered:",arr3.mkString(","))

  /*
    aggregate： 分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的聚合
   */
   val res: Int = rdd1.aggregate(10)(_ + _, _ + _)// [1,2,10] 13   [3,4,10] 17    13 + 17 + 10  = 40
   println(res)

   val res2: Int = rdd1.fold(10)(_ + _)
   println(res2)

  /*
  countByKey: 返回RDD中每个key出现的次数
   */
  val rdd3 = sc.makeRDD( List(  ("a",1), ("a",2), ("b",3), ("b",4),("c",5)  ) )
  val ckRdd: collection.Map[String, Long] = rdd3.countByKey()
  println(ckRdd)

  /*
  save相关算子
   saveAsTextFile：将RDD中的数据保存到文本文件中
   saveAsSequenceFile：将RDD中的数据保存到SequenceFile文件中 数据格式必须是K,V类型
   saveAsObjectFile：将RDD中的数据保存到ObjectFile文件中
   */
  rdd3.saveAsTextFile("output2")
  rdd3.saveAsSequenceFile("output3")
  rdd3.saveAsObjectFile("output4")

  sc.stop()

}
